use ctp_rs::{
    QryInstrumentField, ReqAuthenticateField, ReqUserLoginField, THOST_TE_RESUME_TYPE, TraderApi,
    TraderSpiMsg,
};
use dotenv::dotenv;
use redis::Client as RedisClient;
use serde_json::{Value, json};
use simplelog::{
    ColorChoice, CombinedLogger, Config, ConfigBuilder, LevelFilter, TermLogger, TerminalMode,
    WriteLogger,
};
use std::env;
use std::fs::OpenOptions;
use std::sync::{Arc, mpsc::channel};
use std::thread;
use tokio::sync::mpsc;
use tokio::time::{self, Duration};

fn diff_json_flat(new_v: &Value, old_v: &Value) -> Vec<(String, String, String)> {
    let mut diffs = Vec::new();
    let Some(new_map) = new_v.as_object() else {
        return diffs;
    };
    let Some(old_map) = old_v.as_object() else {
        return diffs;
    };
    for (k, new_val) in new_map.iter() {
        if let Some(old_val) = old_map.get(k) {
            if new_val != old_val {
                diffs.push((k.clone(), old_val.to_string(), new_val.to_string()));
            }
        } else {
            diffs.push((k.clone(), "<missing>".to_string(), new_val.to_string()));
        }
    }
    // 也把旧值里有但新值没有的字段列出来
    for (k, old_val) in old_map.iter() {
        if !new_map.contains_key(k) {
            diffs.push((k.clone(), old_val.to_string(), "<missing>".to_string()));
        }
    }
    diffs
}

fn cleanup_keys(client: &RedisClient) {
    // 清理 Redis 中的 keys：集合和每个 instrument 的 JSON
    log::info!("Cleaning up Redis keys before exit...");
    match client.get_connection() {
        Ok(mut con) => {
            // 删除集合 key
            if let Err(e) = redis::cmd("DEL").arg("instruments").query::<i32>(&mut con) {
                log::error!("Failed to delete set 'instruments': {e}");
            }

            // 删除每个 instrument 下的哈希
            let key = "instrument:*";
            let keys: Vec<String> = match redis::cmd("KEYS").arg(key).query(&mut con) {
                Ok(ks) => ks,
                Err(e) => {
                    log::error!("Failed to fetch keys with pattern {key}: {e}");
                    vec![]
                }
            };

            keys.iter().for_each(|key| {
                if let Err(e) = redis::cmd("DEL").arg(&key).query::<i32>(&mut con) {
                    log::error!("Failed to delete hash {key}: {e}");
                }
            });
            
            // 删除Account信息
            let key = "account:*";
            let keys: Vec<String> = match redis::cmd("KEYS").arg(key).query(&mut con) {
                Ok(ks) => ks,
                Err(e) => {
                    log::error!("Failed to fetch keys with pattern {key}: {e}");
                    vec![]
                }
            };
            keys.iter().for_each(|key| {
                if let Err(e) = redis::cmd("DEL").arg(key).query::<i32>(&mut con) {
                    log::error!("Failed to delete hash {key}: {e}");
                }
            });

            log::info!("Redis keys cleanup finished");
        }
        Err(e) => {
            log::error!("Failed to get Redis connection for cleanup: {e}");
        }
    }
}

// 参数改为从 .env / 环境变量读取

#[tokio::main]
async fn main() {
    // 读取 .env
    dotenv().ok();

    // 从环境变量读取配置
    let broker_id = env::var("BROKER_ID").expect("BROKER_ID not set");
    let user_id = env::var("USER_ID").expect("USER_ID not set");
    let password = env::var("PASSWORD").expect("PASSWORD not set");
    let app_id = env::var("APP_ID").expect("APP_ID not set");
    let auth_code = env::var("AUTH_CODE").expect("AUTH_CODE not set");
    let front_td = env::var("FRONT_TD").expect("FRONT_TD not set");
    let flow_path = env::var("FLOW_PATH").expect("FLOW_PATH not set");
    let redis_url = env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1/".to_string());

    // 初始化日志：
    let file = OpenOptions::new()
        .create(true)
        .append(true) // 日志写入模式
        .open("test.log")
        .expect("无法打开或创建 test.log");

    // 使用本地时间对齐日志时间戳（若获取本地偏移失败则退回默认）
    let mut log_cfg_builder = ConfigBuilder::new();
    let _ = log_cfg_builder.set_time_offset_to_local();
    let log_config: Config = log_cfg_builder.build();

    CombinedLogger::init(vec![
        // 文件：记录所有级别（Trace 及以上）
        WriteLogger::new(LevelFilter::Trace, log_config.clone(), file),
        // 控制台（stdout）：Info 及以上
        TermLogger::new(
            LevelFilter::Info,
            log_config,
            TerminalMode::Stdout,
            ColorChoice::Auto,
        ),
    ])
    .expect("初始化日志失败");
    // SPI 仍使用 std::sync::mpsc 作为回调通道
    let (std_tx, std_rx) = channel::<TraderSpiMsg>();

    // 初始化 Redis 客户端与状态
    let redis_client = Arc::new(RedisClient::open(redis_url).expect("无法创建 Redis 客户端"));
    // 持久连接（避免每条合约都新建连接导致端口耗尽/10048）
    let mut redis_con = redis_client
        .get_connection()
        .expect("无法连接 Redis，请检查 REDIS_URL 或服务状态");

    // 捕获 Ctrl+C，确保退出时清理 keys
    {
        let client = Arc::clone(&redis_client);
        ctrlc::set_handler(move || {
            cleanup_keys(&client);
            std::process::exit(0);
        })
        .expect("设置 Ctrl+C 处理失败");
    }
    let api = Arc::new(TraderApi::CreateTraderApiAndSpi(
        std_tx,
        flow_path.clone(),
        true,
    ));
    api.RegisterFront(front_td);
    api.SubscribePublicTopic(THOST_TE_RESUME_TYPE::THOST_TERT_QUICK as i32);
    // api.SubscribePrivateTopic(THOST_TE_RESUME_TYPE::THOST_TERT_RESTART as i32);
    api.SubscribePrivateTopic(THOST_TE_RESUME_TYPE::THOST_TERT_QUICK as i32);
    api.Init();

    // 桥接线程：把 std_rx -> tokio_rx
    let (tokio_tx, mut tokio_rx) = mpsc::unbounded_channel::<TraderSpiMsg>();
    thread::spawn(move || {
        while let Ok(msg) = std_rx.recv() {
            if tokio_tx.send(msg).is_err() {
                break;
            }
        }
    });

    // 5 秒周期性查询持仓
    let mut interval = time::interval(Duration::from_secs(5));
    let counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));

    loop {
        tokio::select! {
            maybe_msg = tokio_rx.recv() => {
                match maybe_msg {
                    Some(TraderSpiMsg::OnFrontConnected) => {
                        log::info!("front connected");
                        let req = ReqAuthenticateField {
                            BrokerID: broker_id.clone(),
                            UserID: user_id.clone(),
                            AuthCode: auth_code.clone(),
                            AppID: app_id.clone(),
                            ..Default::default()
                        };
                        api.ReqAuthenticate(req, 0);
                    }
                    Some(TraderSpiMsg::OnRspAuthenticate(_, rsp_info, _, _)) => {
                        if rsp_info.ErrorID != 0 {
                            log::error!("auth failed: {:?}", rsp_info);
                            // 退出前清理 Redis keys
                            cleanup_keys(&redis_client);
                            std::process::exit(1);
                        } else {
                            log::info!("auth success: {:?}", rsp_info);

                            let req = ReqUserLoginField {
                                BrokerID: broker_id.clone(),
                                UserID: user_id.clone(),
                                Password: password.clone(),
                                ..Default::default()
                            };
                            api.ReqUserLogin(req, 0);
                        }
                    }
                    Some(TraderSpiMsg::OnRspUserLogin(_, rsp_info, _, _)) => {
                        if rsp_info.ErrorID != 0 {
                            log::error!("user login failed: {:?}", rsp_info);
                            // 退出前清理 Redis keys
                            cleanup_keys(&redis_client);
                            std::process::exit(1);
                        } else {
                            log::info!("user login success: {:?}", rsp_info);
                            let mut instrument = QryInstrumentField::default();
                            // instrument.ProductID = "au".to_string();
                            instrument.InstrumentID = "au2512".to_string();
                            log::info!("qry instrument {instrument:?}"); // 如果为空则返回全部合约
                            api.ReqQryInstrument(instrument, 0);

                            // 测试一个发送报单请求
                            let mut f = ctp_rs::InputOrderField::default();
                            f.BrokerID = broker_id.clone();
                            f.InvestorID = user_id.clone();
                            f.UserID = user_id.clone();
                            // f.ExchangeID = "SHFE".to_string(); // 好像可以不用给定
                            // f.ClientID = 12345.to_string(); // 废弃（没用的，会被覆盖）
                            f.InstrumentID = "au2510".to_string();
                            // 此处注意, ctp是买平和卖平，而不是平多和平空 !!
                            f.Direction = ctp_rs::THOST_FTDC_D_Buy;
                            // ctp原生 如果直接close则自动选择平昨
                            // f.CombOffsetFlag = "0".to_string(); // '0'开仓, '1'平仓, '3'平今, '4'平昨
                            f.CombOffsetFlag = format!("{}", ctp_rs::THOST_FTDC_OF_Close as char);
                            f.CombHedgeFlag = format!("{}", ctp_rs::THOST_FTDC_HF_Speculation as char);

                            // 必须限价单!
                            f.OrderPriceType = ctp_rs::THOST_FTDC_OPT_LimitPrice;
                            // 必须在limit中间
                            f.LimitPrice = 752.38;

                            // AutoSuspend 原生版无影响
                            // f.IsAutoSuspend = 0;

                            // f.OrderPriceType = ctp_rs::THOST_FTDC_OPT_AnyPrice;
                            // f.LimitPrice = 900.00 ; // SimNOW必须填写价格(即便是市价单)
                            // f.IsAutoSuspend = 0;

                            f.VolumeTotalOriginal = 1;
                            f.TimeCondition = ctp_rs::THOST_FTDC_TC_GFD;
                            f.VolumeCondition = ctp_rs::THOST_FTDC_VC_AV;
                            f.MinVolume = 1;

                            // Client-OrderSeq
                            
                            f.OrderRef = "123-456".to_string(); // 需要自行生成唯一值(根据此来做分账等信息)
                            f.ContingentCondition = ctp_rs::THOST_FTDC_CC_Immediately;
                            f.ForceCloseReason = ctp_rs::THOST_FTDC_FCC_NotForceClose;
                            log::info!("Inserting order: {:?}", f);
                            let _ = api.ReqOrderInsert(f, 0);

                        }
                    }
                    Some(TraderSpiMsg::OnRspOrderAction(input_order_action, rsp_info, _, _)) => {
                        log::info!("OrderAction response: {:?}", input_order_action);
                        if rsp_info.ErrorID != 0 {
                            log::error!("order action failed: {:?}", rsp_info);
                        } else {
                            log::info!("order action success: {:?}", rsp_info);
                        }
                    }
                    Some(TraderSpiMsg::OnRspOrderInsert(input_order, rsp_info, _, _)) => {
                        log::info!("OrderInsert response: {:?}", input_order);
                        if rsp_info.ErrorID != 0 {
                            log::error!("order insert failed: {:?}", rsp_info);
                        } else {
                            log::info!("order insert success: {:?}", rsp_info);
                        }
                    }
                    Some(TraderSpiMsg::OnRtnOrder(order)) => {
                        log::info!("Order response: {:?}", order);
                        // 阶段为 受理(97) => 报入(未成交50/51) => 成交(48)
                        // 或者 受理(97) => 成交(48/49)
                        // 未成交会持续回报
                        // 撤单成功在此回报, 53
                        // if order.OrderStatus == ctp_rs::THOST_FTDC_OST_NoTradeQueueing {
                        //     // 发送撤单
                        //     let mut f = ctp_rs::InputOrderActionField::default();
                        //     f.BrokerID = broker_id.clone();
                        //     f.UserID = user_id.clone();
                        //     f.InvestorID = user_id.clone();
                        //     f.ExchangeID = order.ExchangeID.clone();
                        //     f.InstrumentID = order.InstrumentID.clone();
                        //     f.OrderRef = order.OrderRef.clone();
                        //     f.OrderSysID = order.OrderSysID.clone();
                        //     f.ActionFlag = ctp_rs::THOST_FTDC_AF_Delete;
                        //     log::info!("Cancel order: {:?}", f);
                        //     let _ = api.ReqOrderAction(f, 0);
                        // }
                    }
                    Some(TraderSpiMsg::OnRtnTrade(trade)) => {
                        log::info!("Trade response: {:?}", trade);
                    }
                    Some(TraderSpiMsg::OnRspQryInvestorPosition(pos, rsp_info, _, _)) => {
                        if rsp_info.ErrorID != 0 {
                            log::error!("qry position failed: {:?}", rsp_info);
                        } else {
                            log::info!("Position response: {:?}", pos);
                            let key = format!("account:{}:{}", pos.InstrumentID, pos.PosiDirection);
                            let value = serde_json::to_string(&pos).unwrap();
                            if let Err(e) = redis::cmd("SET").arg(&key).arg(value.to_string()).query::<String>(&mut redis_con) {
                                log::error!("Failed to write position to Redis: {e}");
                            }
                        }
                    }
                    Some(TraderSpiMsg::OnRspQryTradingAccount(account, rsp_info, _, _)) => {
                        if rsp_info.ErrorID != 0 {
                            log::error!("qry account failed: {:?}", rsp_info);
                        } else {
                            log::info!("Account response: {:?}", account);
                            let key = format!("account:balance");
                            let value = serde_json::to_string(&account).unwrap();
                            if let Err(e) = redis::cmd("SET").arg(&key).arg(value.to_string()).query::<String>(&mut redis_con) {
                                log::error!("Failed to write account to Redis: {e}");
                            }
                        }
                    }
                    Some(TraderSpiMsg::OnRspQryInstrument(instrument, rsp_info, _, _)) => {
                        if instrument.is_null {
                            log::error!("qry instrument: {:?}", rsp_info);
                            // 退出前清理 Redis keys
                            cleanup_keys(&redis_client);
                            std::process::exit(1);
                        }

                        // 保存到 Redis：集合 'instruments' 及 JSON'
                        if let Err(e) = (|| -> Result<(), redis::RedisError> {
                            let exchange_id = &instrument.ExchangeID;
                            let product_id = &instrument.ProductID;
                            let inst_id = &instrument.InstrumentID;

                            // JSON 写入：包含 InstrumentField 的各字段
                            let key = format!("instrument:{exchange_id}:{product_id}:{inst_id}");
                            let value = serde_json::to_value(&instrument).unwrap();
                            // 如果 key 已存在，则告警并打印变更差异；存在与否决定是否向列表追加
                            let existing: Option<String> = redis::cmd("GET").arg(&key).query(&mut redis_con)?;
                            match existing {
                                Some(old_s) => {
                                    if let Ok(old_json) = serde_json::from_str::<Value>(&old_s) {
                                        let diffs = diff_json_flat(&value, &old_json);
                                        if diffs.is_empty() {
                                            log::debug!("instrument {} already exists with same content", inst_id);
                                        } else {
                                            log::warn!("instrument {} already exists; {} field(s) changed", inst_id, diffs.len());
                                            for (k, ov, nv) in diffs {
                                                log::warn!("  {}: {} -> {}", k, ov, nv);
                                            }
                                        }
                                    } else {
                                        log::warn!("instrument {} exists but old JSON invalid; overwriting", inst_id);
                                    }
                                    // 覆盖为最新内容
                                    let _: () = redis::cmd("SET").arg(&key).arg(value.to_string()).query(&mut redis_con)?;
                                }
                                None => {
                                    // 新 key：先写入，再把 ID 放入集合（set 去重）
                                    let _: () = redis::cmd("SET").arg(&key).arg(value.to_string()).query(&mut redis_con)?;
                                    let _: () = redis::cmd("SADD").arg("instruments").arg(&inst_id).query(&mut redis_con)?;
                                }
                            }
                            Ok(())
                        })() {
                            log::error!("Redis write failed: {e}");
                        }

                        // log::debug!("{:?}", instrument);
                    }
                    _ => {
                        log::debug!("{:?}", maybe_msg);
                    }
                }
            }
        _ = interval.tick() => {
                // 定时查询持仓（全部合约）
                // 偶数次查询持仓，奇数次查询资金(不能同时查询, 需要延时)
                let count = counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                match count % 2 {
                    0 => {
                        let mut pos = ctp_rs::QryInvestorPositionField::default();
                        pos.BrokerID = broker_id.clone();
                        pos.InvestorID = user_id.clone();
                        pos.InstrumentID = "".to_string();
                        pos.ExchangeID = "".to_string();
                        log::info!("Periodic position query");
                        let _ = api.ReqQryInvestorPosition(pos, 0);
                    }
                    _ => {
                        let mut account = ctp_rs::QryTradingAccountField::default();
                        account.BrokerID = broker_id.clone();
                        account.InvestorID = user_id.clone();
                        log::info!("Periodic account query");
                        let _ = api.ReqQryTradingAccount(account, 0);
                    }
                };
            }
        }
    }
}
